package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.DimJoinFunction;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author Felix
 * @date 2024/10/14
 * 发送异步请求进行维度关联的模板类
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {

    StatefulRedisConnection<String, String> asyncRedisConn;
    AsyncConnection asyncHBaseConn;

    @Override
    public void open(Configuration parameters) throws Exception {
        asyncRedisConn = RedisUtil.getRedisAsyncConnection();
        asyncHBaseConn = HBaseUtil.getHBaseAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeRedisAsyncConnection(asyncRedisConn);
        HBaseUtil.closeHBaseAsyncConnection(asyncHBaseConn);
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        CompletableFuture
                //创建异步编排对象  有返回值，这个线程任务的返回值会作为下一个线程任务的参数进行传递
                .supplyAsync(
                        new Supplier<JSONObject>() {
                            @Override
                            public JSONObject get() {
                                //以异步的方式从Redis中获取维度数据
                                JSONObject dimJsonObj = RedisUtil.readDimAsync(asyncRedisConn, getTableName(), getRowKey(obj));
                                return dimJsonObj;
                            }
                        }
                )
                //创建线程任务  有入参，有返回值
                .thenApplyAsync(
                        new Function<JSONObject, JSONObject>() {
                            @Override
                            public JSONObject apply(JSONObject dimJsonObj) {
                                if (dimJsonObj != null) {
                                    //如果在Redis中获取到了对应的维度，直接将其作为返回值进行返回(缓存命中)
                                    System.out.println("~~~从Redis中获取" + getTableName() + "表的维度" + getRowKey(obj) + "数据~~~");
                                }else{
                                    //如果在Redis中没有获取到对应的维度，以异步的方式到HBase中查询
                                    dimJsonObj = HBaseUtil.readDimAsync(asyncHBaseConn, Constant.HBASE_NAMESPACE,getTableName(),getRowKey(obj));
                                    if(dimJsonObj != null){
                                        //以异步的方式将查询到维度数据放到Redis中进行缓存，方便下次查询使用
                                        System.out.println("~~~从HBase中获取" + getTableName() + "表的维度" + getRowKey(obj) + "数据~~~");
                                        RedisUtil.writeDimAsync(asyncRedisConn,getTableName(),getRowKey(obj),dimJsonObj);
                                    }else{
                                        System.out.println("~~~没有找到要关联的维度~~~");
                                    }
                                }
                                return dimJsonObj;
                            }
                        }
                )
                //创建线程任务   有入参，无返回值
                .thenAcceptAsync(
                        new Consumer<JSONObject>() {
                            @Override
                            public void accept(JSONObject dimJsonObj) {
                                if (dimJsonObj != null){
                                    //将维度属性补充到流中对象上
                                    addDims(obj,dimJsonObj);
                                }
                                //获取数据库交互的结果并发送给 ResultFuture 的 回调 函数
                                resultFuture.complete(Collections.singleton(obj));
                            }
                        }
                );
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        throw new RuntimeException("请检查HBase维度表历史数据是否同步\n" +
                "检查zk、kafka、maxwell、hdfs、hbase、redis、dwd下单事实表处理应用是否正常启动\n" +
                "请联系昌平彦祖");
    }
}
